package com.bw.dws;

import com.bw.bean.DeviceData;
import com.bw.bean.UserBehavior;
import com.bw.bean.UserProfileTag;
import com.bw.dwd.ProfileSource;
import com.bw.util.ProfileCalcUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * 达摩盘基础特征Flink实时作业主程序
 * 工单编号：大数据-用户画像-11-达摩盘基础特征（文档🔶1-10）
 */
public class ProfileFlinkJob {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置Checkpoint（HDFS存储，保障状态安全）
        env.enableCheckpointing(300000); // 5分钟一次
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hdfs01:8020/flink/checkpoints/profile_job");

        // 2. 接入Source流
        // （1）用户行为流→年龄/性别标签
        DataStream<UserBehavior> behaviorStream = ProfileSource.getUserBehaviorSource(env);

        // （2）设备数据流→身高/体重标签
        DataStream<DeviceData> deviceStream = ProfileSource.getDeviceDataSource(env);

        // （3）用户基础信息CDC流→星座/出生年代标签
        DataStream<Map<String, Object>> baseInfoStream = ProfileSource.getUserBaseCDCSource(env)
                .map(jsonStr -> {
                    Map<String, Object> baseMap = new HashMap<>();
                    Map<String, Object> data = JSON.parseObject(jsonStr, Map.class);
                    Map<String, Object> after = (Map<String, Object>) data.get("after");
                    if (after == null) return baseMap;

                    String userId = (String) after.get("user_id");
                    String birthDate = (String) after.get("birth_date");
                    // 计算星座和出生年代（文档🔶1-251/1-17）
                    String constellation = ProfileCalcUtil.calcConstellation(birthDate);
                    String birthEra = ProfileCalcUtil.calcBirthEra(birthDate);

                    baseMap.put("userId", userId);
                    baseMap.put("constellation", constellation);
                    baseMap.put("birthEra", birthEra);
                    baseMap.put("updateTime", new java.sql.Timestamp(System.currentTimeMillis()));
                    return baseMap;
                })
                .filter(map -> !map.isEmpty())
                .keyBy(map -> (String) map.get("userId"));

        // 3. 执行Transform：计算各标签
        // （1）年龄标签
        DataStream<Map<String, Object>> ageTagStream = behaviorStream
                .keyBy(UserBehavior::getUserId)
                .process(new AgeTagCalc());

        // （2）性别标签
        DataStream<Map<String, Object>> genderTagStream = behaviorStream
                .keyBy(UserBehavior::getUserId)
                .process(new GenderTagCalc());

        // （3）身高体重标签
        DataStream<Map<String, Object>> bodyTagStream = deviceStream
                .keyBy(DeviceData::getUserId)
                .process(new BodyTagCalc());

        // 4. 流合并：关联所有标签（按用户ID+更新时间）
        DataStream<UserProfileTag> finalTagStream = ageTagStream
                .connect(genderTagStream)
                .keyBy(
                        map -> (String) map.get("userId"),
                        map -> (String) map.get("userId")
                )
                .process(new KeyedCoProcessFunction<String, Map<String, Object>, Map<String, Object>, UserProfileTag>() {
                    // 状态存储各标签结果
                    private ValueState<Map<String, Object>> ageState;
                    private ValueState<Map<String, Object>> genderState;
                    private ValueState<Map<String, Object>> bodyState;
                    private ValueState<Map<String, Object>> baseState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ageState = getRuntimeContext().getState(new ValueStateDescriptor<>("ageState", 
                                org.apache.flink.api.common.typeinfo.Types.MAP(
                                    org.apache.flink.api.common.typeinfo.Types.STRING,
                                    org.apache.flink.api.common.typeinfo.Types.GENERIC(Object.class)
                                )));
                        genderState = getRuntimeContext().getState(new ValueStateDescriptor<>("genderState", 
                                org.apache.flink.api.common.typeinfo.Types.MAP(
                                    org.apache.flink.api.common.typeinfo.Types.STRING,
                                    org.apache.flink.api.common.typeinfo.Types.GENERIC(Object.class)
                                )));
                        bodyState = getRuntimeContext().getState(new ValueStateDescriptor<>("bodyState", 
                                org.apache.flink.api.common.typeinfo.Types.MAP(
                                    org.apache.flink.api.common.typeinfo.Types.STRING,
                                    org.apache.flink.api.common.typeinfo.Types.GENERIC(Object.class)
                                )));
                        baseState = getRuntimeContext().getState(new ValueStateDescriptor<>("baseState", 
                                org.apache.flink.api.common.typeinfo.Types.MAP(
                                    org.apache.flink.api.common.typeinfo.Types.STRING,
                                    org.apache.flink.api.common.typeinfo.Types.GENERIC(Object.class)
                                )));
                    }

                    // 合并各标签状态，生成最终结果
                    @Override
                    public void processElement1(Map<String, Object> ageMap, Context ctx, Collector<UserProfileTag> out) throws Exception {
                        ageState.update(ageMap);
                        emitFinalTag(out);
                    }

                    @Override
                    public void processElement2(Map<String, Object> genderMap, Context ctx, Collector<UserProfileTag> out) throws Exception {
                        genderState.update(genderMap);
                        emitFinalTag(out);
                    }

                    // 生成最终标签对象
                    private void emitFinalTag(Collector<UserProfileTag> out) throws Exception {
                        Map<String, Object> age = ageState.value();
                        Map<String, Object> gender = genderState.value();
                        Map<String, Object> body = bodyState.value();
                        Map<String, Object> base = baseState.value();

                        if (age == null || gender == null || body == null || base == null) return;

                        UserProfileTag tag = new UserProfileTag();
                        tag.setUserId((String) age.get("userId"));
                        tag.setAgeGroup((String) age.get("ageGroup"));
                        tag.setGenderTag((String) gender.get("genderTag"));
                        tag.setHeight((Double) body.getOrDefault("height", null));
                        tag.setWeight((Double) body.getOrDefault("weight", null));
                        tag.setBirthEra((String) base.get("birthEra"));
                        tag.setConstellation((String) base.get("constellation"));
                        tag.setUpdateTime((java.sql.Timestamp) age.get("updateTime"));

                        out.collect(tag);
                    }
                });

        // 5. 写入Sink
        finalTagStream.addSink(new ProfileSink()).name("ProfileSink");

        // 6. 执行作业
        env.execute("DamoPanProfileFlinkJob（工单：大数据-用户画像-11-达摩盘基础特征）");
    }
}